package com.hhf.rrd.usage.ck;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

/**
 * ck
 *      开启socket端口：nc -l 9999
 *      查看端口占用: lsof -i :9999
 *
 * @author huanghaifeng15
 * @date 2022/2/10 11:09
 **/
public class CheckPointApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 开启CK
        env.enableCheckpointing(5000);

        // 确保一次性语义
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);

        // 是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // 设置StateBackend
        env.setStateBackend(new FsStateBackend("file:///Users/huanghaifeng15/Desktop/Flink/workspace/flink-learning/checkpoints"));

        // env.setStateBackend(new FsStateBackend("hdfs://Localhost:8020/flink-learning-checkpoints"));

        // 自定义重启策略 (间隔5秒，尝试重启3次)
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,
                Time.of(5, TimeUnit.SECONDS)));

        DataStreamSource<String> source = env.socketTextStream("localhost", 9999);
        source.map(new MapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        if(value.contains("dean")) {
                            throw new RuntimeException("dean来了，快跑..");
                        } else {
                            return value.toLowerCase();
                        }
                    }
                }).flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String value, Collector<String> out) throws Exception {
                        String[] splits = value.split(",");
                        for (String split : splits) {
                            out.collect(split);
                        }
                    }
                }).map(new MapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String value) throws Exception {
                        return Tuple2.of(value, 1);
                    }
                }).keyBy(x -> x.f0)
                .sum(1)
                .print();

        env.execute();
    }
}
